package com.gitee.checkpoint

import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment,_}

object CheckpointSet {
  def main(args: Array[String]): Unit = {
    //1.准备环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //2.设置参数
    //=============================必须设置==================================
    //设置State和Checkpoint的保存后端(默认是保存在内存中,开发中一般为HDFS)
    if(args.length<1){
      env.setStateBackend(new FsStateBackend("file:///C:\\develop\\code\\idea\\FlinkStudy\\FlinkDemo\\ckp"))
    }else{
      env.setStateBackend(new FsStateBackend(args(0)))
      //env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink-checkpoint/checkpoint"))
    }
    //设置Checkpointing时间间隔为1000ms,意思是做 2 个 Checkpoint 的事件间隔为1000ms。Checkpoint 做的越频繁，恢复数据时就越简单，同时 Checkpoint 相应的也会有一些IO消耗。
    env.enableCheckpointing(1000)//(默认情况下checkpoint是没有开启的)
    //=============================必须设置==================================

    //=============================建议设置=================================
    //TODO 设置如果在做Checkpoint过程中出现错误，是否让整体任务失败：true是  false不是
    //默认值是true,开发中大多数情况下应设置为false,保证Checkpoint失败不让任务失败
    env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
    //TODO 设置是否清理检查点,表示 Cancel时是否需要保留当前的 Checkpoint，默认 Checkpoint会在作业被Cancel时被删除
    //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：true,当作业被取消时，删除外部的checkpoint(默认值)
    //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION：false,当作业被取消时，保留外部的checkpoint
    //默认值是DELETE_ON_CANCELLATION即true,表示任务取消时删除Checkpoint
    //开发中应该设置为RETAIN_ON_CANCELLATION即false,表示任务取消时保留Checkpoint,便于重启时恢复
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    //设置重启策略
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 5000))
    //TODO 设置两个Checkpoint 之间最少等待时间,如设置Checkpoint之间最少是要等 500ms，也就是刚做完一个 Checkpoint。比如某个 Checkpoint 做了700ms，按照原则过 300ms 应该是做下一个 Checkpoint，因为设置了 1000ms 做一次 Checkpoint 的，但是中间的等待时间比较短，不足 500ms 了，需要多等 200ms，因此以这样的方式防止 Checkpoint 太过于频繁而导致业务处理的速度下降。
    //默认值0
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    //=============================建议设置=================================

    //=============================一般使用默认值即可=================================
    //设置checkpoint的执行模式为EXACTLY_ONCE表示精准一次/精确一次/就是数据只会被处理一次,
    //TODO 底层由2PC(两阶段提交协议)保证,有一定的性能损耗,如果对数据精准性要求不高,对性能要求很高,可以设置为AT_LEAST_ONCE(至少一次,有可能会重复计算)
    //默认值就是EXACTLY_ONCE
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //设置checkpoint的超时时间,如果 Checkpoint在 60s内尚未完成说明该次Checkpoint失败,则丢弃。
    //默认10min
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    //设置同一时间有多少个checkpoint可以同时执行
    //默认值就是1
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    //=============================一般使用默认值即可=================================

    //3.获取数据
    val lines: DataStream[String] =  env.addSource(new MyStringSourceScala)
    //4.处理数据
    val result = lines.flatMap(_.split(" ")).map(_->1).keyBy(0).sum(1)
    //5.输出结果
    result.print()
    //6.启动执行
    env.execute()


    //注意:上面只是将State每隔x秒进行一次Checkpoint持久化存储
    //当手动停止任务并启动任务并不会读取状态信息接着计算
    //如果要实现下次重启可以接着上次的重新计算应该要使用后面学习的Savepoint机制,手动指定从哪个一个Checkpoint目录读取数据并接着计算
    //或者使用后面学习的重启策略可以实现自动从Checkpoint目录读取数据并接着计算

  }

  class MyStringSourceScala extends SourceFunction[String] {
    var isRunning = true
    override def run(ctx: SourceContext[String]) = {
      while (isRunning) {
        ctx.collect("hello " + "haha")
        TimeUnit.SECONDS.sleep(1)
      }
    }
    override def cancel() = {
      isRunning = false
    }
  }

}
